package net.opentsdb.plugin.impl;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.stumbleupon.async.Callback;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.TSDB;
import net.opentsdb.core.TsuidExchange;
import net.opentsdb.meta.MetaDataCache;
import net.opentsdb.plugin.listener.TsuidExpiringListener;
import net.opentsdb.stats.StatsCollector;
import net.opentsdb.utils.TsdbPair;
import net.opentsdb.utils.enums.FamQuls;
import net.opentsdb.utils.enums.TsuidChecks;
import net.opentsdb.utils.expiringmap.ExpirationPolicy;
import net.opentsdb.utils.expiringmap.ExpiringMap;
import net.opentsdb.utils.hbase.HbaseMetaUtils;
import net.opentsdb.utils.hbase.HbaseTagUtils;
import org.hbase.async.KeyValue;
import org.hbase.async.Scanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * 自定义实现local meta cache
 * 即本地存储时间线
 *
 */
public class LocalMetaCache extends MetaDataCache {

    private static final Logger LOG = LoggerFactory.getLogger(LocalMetaCache.class);

    public final static TimeUnit TTL_UNIT = TimeUnit.DAYS;

    private static long hbaseDuration;
    private static long tsuidMaxEntries;
    private static int tsuidTTL;
    private static HbaseMetaUtils hbaseMetaUtils;
    private static HbaseTagUtils hbaseTagUtils;
    private TSDB tsdb;

    //记录所有的tsuid
    private Cache<Integer, Boolean> metaCache;
    private ExpiringMap<ByteBuffer, LongAdder> metaMap;
    //统计总数
    private Cache<String, LongAdder> countCache;

    private final static String TSUID_COUNT_KEY = "TSUID_COUNT_KEY";
    private final static String DATAPOINT_COUNT_KEY = "DATAPOINT_COUNT_KEY";

    @Override
    public void initialize(final TSDB tsdb) {
        this.tsdb = tsdb;
        this.tsuidMaxEntries = tsdb.getConfig().tsuid_max_entries();
        this.tsuidTTL = tsdb.getConfig().tsuid_ttl();
        this.metaCache = CacheBuilder.newBuilder().maximumSize(tsuidMaxEntries).build();
        this.countCache = CacheBuilder.newBuilder().maximumSize(2).build();
        this.hbaseMetaUtils = tsdb.getHbaseWrapper().metaUtils;
        this.hbaseTagUtils = tsdb.getHbaseWrapper().tagUtils;
        hbaseDuration = TimeUnit.MILLISECONDS.convert(5,TimeUnit.MINUTES); // TODO config中获取

        metaMap = ExpiringMap
                .builder()
                .maxSize((int)tsuidMaxEntries)
                .variableExpiration()
                .ttl(tsuidTTL,TTL_UNIT)
//                .expirationPolicy(ExpirationPolicy.ACCESSED) //只要有访问动作，过期时间就会被延后
                .expirationPolicy(ExpirationPolicy.CREATED) //ACCESSED模式 map get操作会重设时间戳
                .build();

        metaMap.addExpirationListener(new TsuidExpiringListener(tsdb));

        // 初始化两个计数器
        countCache.put(TSUID_COUNT_KEY,new LongAdder());
        countCache.put(DATAPOINT_COUNT_KEY,new LongAdder());

        loadMeta();
    }

    /**
     * 加载meta数据信息到localCache
     *
     */
    private void loadMeta(){
        LongAdder size = new LongAdder();

        Scanner scanner = tsdb.getClient().newScanner(tsdb.metaTable());
        scanner.setFamily(FamQuls.META.FAMILY.NAME.getByte());
        // 每次最多8*1024笔数据
        scanner.setMaxNumRows(8192);

        final class ScannerCB implements Callback<Object,
                        ArrayList<ArrayList<KeyValue>>> {

            public Object scan() {
                return scanner.nextRows().addCallback(this);
            }

            @Override
            public Object call(final ArrayList<ArrayList<KeyValue>> rows) throws Exception {
                try {
                    if (rows == null) {
                        //同步hbase计数
                        hbaseMetaUtils.initTSMetaCount(size.sum());
                        LOG.debug("async load meta end..., metaCache size:{}, hbase size:{}",metaMap.size(),size);
                        return null;
                    }

                    for (final ArrayList<KeyValue> arr : rows) {
                        for(final KeyValue row : arr){
                            ByteBuffer buf = ByteBuffer.wrap(row.key());
//                            metaMap.putIfAbsent(buf,new LongAdder(), row.timestamp(), 0);
                            metaCache.put(buf.hashCode(), true);
                        }
                        size.add(arr.size());
                    }
                    return scan();
                } catch (Exception e) {
                    LOG.error("init meta data error:",e);
                    return null;
                }
            }
        }

        new ScannerCB().scan();

        LOG.debug(" load meta start.......");

        // 减去计数器 目前没用处
//        countCache.getIfPresent(TSUID_COUNT_KEY).reset();
//        countCache.getIfPresent(TSUID_COUNT_KEY).add(metaMap.size());
//        countCache.getIfPresent(DATAPOINT_COUNT_KEY).add(metaMap.size());

        //同步hbase计数
//        hbaseMetaUtils.initTSMetaCount(size.sum());

        // 决定是否需要初始化计数
        /*if(hbaseMetaUtils.getTsuidCount() == null){
            hbaseMetaUtils.initTSMetaCount(countCache.getIfPresent(TSUID_COUNT_KEY).sum());
        }*/
    }

    @Override
    public Deferred<Object> shutdown() {
        if(metaCache != null){
            metaCache.invalidateAll();
        }
        if(countCache != null){
            countCache.invalidateAll();
        }
        LOG.info("local meta cache invalidate all tsuid");
        return Deferred.fromResult(null);
    }

    @Override
    public String version() {
        return "0.0.1";
    }

    @Override
    public void collectStats(StatsCollector collector) {
        // null
    }

    /**
     * 来一笔新的数据
     *
     * @param tsuid The tsuid to increment
     */
    @Override
    public void increment(byte[] tsuid, TsuidChecks checks, long baseTime) {
        ByteBuffer buffer = ByteBuffer.wrap(tsuid);

//        LOG.debug("tsuid:{} check name:{}  map size:{}",UniqueId.uidToString(tsuid),checks.name(),metaMap.size());
        switch (checks){
            //本地已存在tsuid
            case LocalTsuidExists:
                // 重设时间
//                metaMap.setTimestramp(buffer);
                metaMap.setTimestramp(buffer, baseTime, TimeUnit.MILLISECONDS);
                //记录本机写入点数
                metaMap.get(buffer).increment();
                // 决定是否更新hbase时间戳
                if(metaMap.getLastTimestramp(buffer)<= System.currentTimeMillis()-hbaseDuration){
                    hbaseMetaUtils.storeCountTSMeta(tsuid, 0);
                    // 设置最后更新时间
                    metaMap.setLastTimestramp(buffer);
                }
                break;
            //hbase已存在tsuid
            case HbaseTsuidExists:
                LongAdder adder = new LongAdder();
                adder.increment();
                metaMap.put(buffer, adder, baseTime, 0L);
                metaCache.put(buffer.hashCode(), true);
                // 更新tsuid时间
                hbaseMetaUtils.storeCountTSMeta(tsuid, 0);
                break;
            case LocalCountFullCapacity:
                break;
            case HbaseCountFullCapacity:
                break;
            case NewInputTsuid:
                LongAdder newAdder = new LongAdder();
                newAdder.increment();
                metaMap.put(buffer, newAdder, baseTime, 0L);
                metaCache.put(buffer.hashCode(), true);

                countCache.getIfPresent(TSUID_COUNT_KEY).increment();
                countCache.getIfPresent(DATAPOINT_COUNT_KEY).increment();

                // 新增时间线、计数加一
                // 通过简单cas实现多实例计数
                if(hbaseMetaUtils.storeCasCountTSMeta(tsuid, 0L))
                    hbaseMetaUtils.incrementDefaultAndGetCounter();

                //保存tag信息
                TsdbPair<String,Map<String,String>> tsdbPair = TsuidExchange.getMetricAndTagFromTSUID(tsdb, tsuid);
                hbaseTagUtils.storeTagTree(tsdbPair.getKey(), tsdbPair.getValue());
                break;
            default:
                break;
        }
    }

    /**
     * 判断能否写入数据，注意这里会同步计数信息
     *
     * @param tsuid
     * @return
     */
    @Override
    public TsuidChecks checks(byte[] tsuid){

        ByteBuffer buffer = ByteBuffer.wrap(tsuid);

        // 本地验证时间线已存在
//        if(metaMap.get(buffer) != null)
        if(metaCache.getIfPresent(buffer.hashCode()) != null)
            return TsuidChecks.LocalTsuidExists;

        // 本地时间线已满
        if(metaCache.size() >= tsuidMaxEntries){
            return TsuidChecks.LocalCountFullCapacity;
        }

        boolean existsHbaseMeta = hbaseMetaUtils.existsTsuid(tsuid);

        // hbase meta已存在该时间线
        if(existsHbaseMeta){
            LongAdder adder = new LongAdder();
            adder.increment();
            metaMap.put(buffer, adder);
            metaCache.put(buffer.hashCode(), true);
            return TsuidChecks.HbaseTsuidExists;
        }else {
            // 新时间线

            Long hbaseMetaCount = hbaseMetaUtils.getTsuidCount();
            hbaseMetaCount = hbaseMetaCount!=null?hbaseMetaCount:0;

            // 同步计数
            countCache.getIfPresent(TSUID_COUNT_KEY).reset();
            countCache.getIfPresent(TSUID_COUNT_KEY).add(hbaseMetaCount);

            // hbase时间线计数已满
            if(hbaseMetaCount >= tsuidMaxEntries) {
                // 将时间线同步
                LOG.debug(" HbaseCountFullCapacity and begin load tsuid to local cache......");
                loadMeta();
                return TsuidChecks.HbaseCountFullCapacity;
            } else return TsuidChecks.NewInputTsuid;
        }
    }

    @Override
    public boolean removeIncrementTsuid(byte[] tsuid) {
        ByteBuffer buffer = ByteBuffer.wrap(tsuid);
//        LOG.debug("start remove local tsuid {}, map size {}, tsuid count is {}",
//                tsuid, metaMap.size(), countCache.getIfPresent(TSUID_COUNT_KEY));

        if(metaMap.containsKey(buffer)){
            metaMap.remove(buffer);
            metaCache.invalidate(buffer.hashCode());
            countCache.getIfPresent(TSUID_COUNT_KEY).decrement();
            countCache.getIfPresent(DATAPOINT_COUNT_KEY).decrement();
//            LOG.debug("remove local tsuid {} successful, current tsuid count is {}",
//                    tsuid, countCache.getIfPresent(TSUID_COUNT_KEY));
            return true;
        }
        return false;
    }

    @Override
    public boolean resetTTL(long ttl) {
        LOG.debug("reset ttl change from {} to {}",metaMap.getTtl(),ttl);
        metaMap.setTtl(ttl, TTL_UNIT);
        return true;
    }

    @Override
    public void decrementCount(){
//        LOG.debug("force decrement map count");
        countCache.getIfPresent(TSUID_COUNT_KEY).decrement();
        countCache.getIfPresent(DATAPOINT_COUNT_KEY).decrement();
    }

}
